package a_o2odata_deal.utils.zhibo_qy

import java.text.SimpleDateFormat
import java.util.Date

import a_aa_amainpackage.a_o2odata_deal.config.config._
import com.alibaba.fastjson.{JSON, JSONArray}
import org.apache.spark.sql.{DataFrame, SQLContext}

/**
  * @ Auther: o2o-rd-0008
  * @ Date:   2020/2/11 14:49
  * @ Param:  ${PARAM}
  * @ Description: 将关联完品牌的数据进行直播处理，将处理结果进行保存，供之后计算其他维度
  */
object handle_zhibo_good {

  def handle_qiye_good(sqlContext:SQLContext):Unit={
    handle_qiye_tmall_good(sqlContext)
    handle_qiye_taobao_good(sqlContext)
  }

  def handle_qiye_tmall_good(sqlContext:SQLContext):Unit={
    sqlContext.udf.register("tranTimeToString",tranTimeToString _)
    sqlContext.udf.register("handle_array",handle_array _)

    val zhibo_data = handle_xiajia(sqlContext)

    zhibo_data.createOrReplaceTempView("zhibo_data")
    //加载直播数据
    //关联直播
    val data_detail = sqlContext.read.json(s"${a_aa_amainpackage.a_o2odata_deal.config.config.tmall_live_detail_path}")
      .dropDuplicates("good_id","liveId")
    val data_id = sqlContext.read.json(s"${a_aa_amainpackage.a_o2odata_deal.config.config.tmall_live_id_path}")
    //.where("cast(liveTime as long)<1575129600 and cast(liveTime as long)>=1572537600")

    data_id.createOrReplaceTempView("data_id")
    data_detail.createOrReplaceTempView("data_detail")

    //店铺主播当期所有直播场次观看人数的累加，数据处理新增字段：“店铺直播观看人数”
    /* sqlContext.sql(
       s"""
          |select
          |*,
          |sum(viewerCount ) over(partition by anchorId) as shopLive_viewCount
          |from data_id
        """.stripMargin).createOrReplaceTempView("data_id")*/

    //首先统计出单品在多少个直播间出现过,1对多的关系，关联出来数据会增多
    val source_data = sqlContext.sql(
      s"""
         |select
         |t1.*,
         |t2.livePriceText,
         |t2.good_id as good_ids
         |from data_id t1
         |inner join
         |data_detail t2
         |on t1.liveId=t2.liveId
       """.stripMargin).drop("good_id").withColumnRenamed("good_ids","good_id").where("good_id!='-1'")

    source_data.createOrReplaceTempView("source_data")

    sqlContext.sql(
      s"""
         |select
         |good_id,
         |anchorId,
         |livePriceText,
         |roomType,
         |liveId,
         |liveTime,
         |roomTypeName,
         |nick,
         |liveTitle,
         |commodityCount,
         |cast(anchorFans as bigint) as anchorFans,
         |--直播次数
         |count(liveId) over(partition by good_id) as liveCount,
         |--观看人数,基于商品的
         |sum(cast(viewerCount as bigint)) over(partition by good_id) as viewcount,
         |--直播均价
         |avg(cast(livePriceText as double)) over(partition by good_id) as liveAvgPriceText
         |from source_data
       """.stripMargin).dropDuplicates("good_id").createOrReplaceTempView("value")


    //每个商品下钻分析（新增）：每条商品每次直播的主播名称、观看人数、直播时间
    sqlContext.sql(
      s"""
         |select distinct
         |good_id,
         |collect_list(liveId) as liveId_collect,
         |collect_list(anchorId) as anchorId_collect,
         |collect_list(nick) as nick_collect,
         |collect_list(viewerCount) as viewerCount_collect,
         |collect_list(tranTimeToString(liveTime)) as liveTime_collect
         |from source_data
         |group by good_id
       """.stripMargin).createOrReplaceTempView("tmp")

    sqlContext.sql(
      s"""
         |select
         |good_id,
         |handle_array(liveId_collect,anchorId_collect,nick_collect,viewerCount_collect,liveTime_collect)
         |as add_to_field
         |from tmp
       """.stripMargin).createOrReplaceTempView("mid")

    //打上add_to_field标签/*
    sqlContext.sql(
      s"""
         |select
         |t1.*,
         |t2.add_to_field
         |from value t1
         |inner join
         |mid t2
         |on t1.good_id=t2.good_id
       """.stripMargin).createOrReplaceTempView("value1")


    //打上是否是自主直播标签
    sqlContext.sql(
      s"""
         |select
         |t1.*,
         |case
         |when t2.userId is null then "false"
         |else "true"
         |end as is_selfLive
         |from value1 t1
         |left join
         |zhibo_data t2
         |on t1.anchorId=t2.userId
       """.stripMargin).dropDuplicates("good_id").createOrReplaceTempView("data")


    val result = sqlContext.sql(
      s"""
         |select
         |t1.*,
         |--t2.anchorId,
         |--t2.livePriceText,
         |--t2.roomType,
         |--t2.liveId,
         |--t2.liveTime,
         |--t2.roomTypeName,
         |--t2.nick,
         |--t2.liveTitle,
         |--t2.commodityCount,
         |--cast(t2.anchorFans as bigint) as anchorFans,
         |case when t2.good_id is null then "" else cast(t2.liveCount as bigint) end as liveCount,
         |case when t2.good_id is null then "" else cast(t2.viewcount as bigint) end as viewcount,
         |case when t2.good_id is null then "" else cast(t2.liveAvgPriceText as double) end as liveAvgPriceText,
         |--t2.add_to_field,
         |case when t2.good_id is null then "" else t2.is_selfLive end as is_selfLive,
         |case when t2.good_id is null then "false" else "true"  end as is_showLive
         |from zhibo_data t1
         |left join
         |data t2
         |on t1.good_id=t2.good_id
       """.stripMargin)
    result.printSchema()
    import org.apache.spark.sql.functions._
    result.agg(count("*"),sum("sellCount"),sum("salesAmount")).show(false)
    result.withColumnRenamed("liveAvgPriceText","liveAvg_priceText")
      .repartition(12).write.orc(s"s3a://o2o-dataproces-group/panzonghao/tmall/${years}/${months}/zhibo_finally")
  }

  def handle_qiye_taobao_good(sqlContext:SQLContext):Unit={
    sqlContext.udf.register("tranTimeToString",tranTimeToString _)
    sqlContext.udf.register("handle_array",handle_array _)

    //val zhibo_data = sqlContext.read.orc(s"${source_result}_modify_xiajia")
    //val zhibo_data = sqlContext.read.orc(s"s3a://dws-data/g_data/${years}/${months}/taobao")
    val zhibo_data = sqlContext.read.orc(s"s3a://o2o-dataproces-group/yang_songjian/product/taobao/2020/3/taobaoFinalSellDatas")

    zhibo_data.createOrReplaceTempView("zhibo_data")
    //加载直播数据
    //关联直播
    val data_detail = sqlContext.read.json(s"${a_aa_amainpackage.a_o2odata_deal.config.config.tmall_live_detail_path}")
      .dropDuplicates("good_id","liveId")
    val data_id = sqlContext.read.json(s"${a_aa_amainpackage.a_o2odata_deal.config.config.tmall_live_id_path}")
    //.where("cast(liveTime as long)<1575129600 and cast(liveTime as long)>=1572537600")

    data_id.createOrReplaceTempView("data_id")
    data_detail.createOrReplaceTempView("data_detail")


    //首先统计出单品在多少个直播间出现过,1对多的关系，关联出来数据会增多
    val source_data = sqlContext.sql(
      s"""
         |select
         |t1.*,
         |t2.livePriceText,
         |t2.good_id as good_ids
         |from data_id t1
         |inner join
         |data_detail t2
         |on t1.liveId=t2.liveId
       """.stripMargin).drop("good_id").withColumnRenamed("good_ids","good_id").where("good_id!='-1'")

    source_data.createOrReplaceTempView("source_data")

    sqlContext.sql(
      s"""
         |select
         |good_id,
         |anchorId,
         |livePriceText,
         |roomType,
         |liveId,
         |liveTime,
         |roomTypeName,
         |nick,
         |liveTitle,
         |commodityCount,
         |cast(anchorFans as bigint) as anchorFans,
         |--直播次数
         |count(liveId) over(partition by good_id) as liveCount,
         |--观看人数,基于商品的
         |sum(cast(viewerCount as bigint)) over(partition by good_id) as viewcount,
         |--直播均价
         |avg(cast(livePriceText as double)) over(partition by good_id) as liveAvgPriceText
         |from source_data
       """.stripMargin).dropDuplicates("good_id").createOrReplaceTempView("value")


    //每个商品下钻分析（新增）：每条商品每次直播的主播名称、观看人数、直播时间
    sqlContext.sql(
      s"""
         |select distinct
         |good_id,
         |collect_list(liveId) as liveId_collect,
         |collect_list(anchorId) as anchorId_collect,
         |collect_list(nick) as nick_collect,
         |collect_list(viewerCount) as viewerCount_collect,
         |collect_list(tranTimeToString(liveTime)) as liveTime_collect
         |from source_data
         |group by good_id
       """.stripMargin).createOrReplaceTempView("tmp")

    sqlContext.sql(
      s"""
         |select
         |good_id,
         |handle_array(liveId_collect,anchorId_collect,nick_collect,viewerCount_collect,liveTime_collect) as add_to_field
         |from tmp
       """.stripMargin).createOrReplaceTempView("mid")

    //打上add_to_field标签/*
    sqlContext.sql(
      s"""
         |select
         |t1.*,
         |t2.add_to_field
         |from value t1
         |inner join
         |mid t2
         |on t1.good_id=t2.good_id
       """.stripMargin).createOrReplaceTempView("value1")


    //打上是否是自主直播标签
    sqlContext.sql(
      s"""
         |select
         |t1.*,
         |case
         |when t2.userId is null then "false"
         |else "true"
         |end as is_selfLive
         |from value1 t1
         |left join
         |zhibo_data t2
         |on t1.anchorId=t2.userId
       """.stripMargin).dropDuplicates("good_id").createOrReplaceTempView("data")


    val result = sqlContext.sql(
      s"""
         |select
         |t1.*,
         |--t2.anchorId,
         |--t2.livePriceText,
         |--t2.roomType,
         |--t2.liveId,
         |--t2.liveTime,
         |--t2.roomTypeName,
         |--t2.nick,
         |--t2.liveTitle,
         |--t2.commodityCount,
         |--cast(t2.anchorFans as bigint) as anchorFans,
         |case when t2.good_id is null then "" else cast(t2.liveCount as bigint) end as liveCount,
         |case when t2.good_id is null then "" else cast(t2.viewcount as bigint) end as viewcount,
         |case when t2.good_id is null then "" else cast(t2.liveAvgPriceText as double) end as liveAvgPriceText,
         |--t2.add_to_field,
         |case when t2.good_id is null then "" else t2.is_selfLive end as is_selfLive,
         |case when t2.good_id is null then "false" else "true"  end as is_showLive
         |from zhibo_data t1
         |left join
         |data t2
         |on t1.good_id=t2.good_id
       """.stripMargin)
    result.printSchema()
    import org.apache.spark.sql.functions._
    result.agg(count("*"),sum("sellCount"),sum("salesAmount")).show(false)
    result.withColumnRenamed("liveAvgPriceText","liveAvg_priceText")
      .repartition(12).write.orc(s"s3a://o2o-dataproces-group/panzonghao/taobao/${years}/${months}/zhibo_finally")

  }

  def handle_xiajia(sqlContext:SQLContext):DataFrame={
    sqlContext.udf.register("min_price",min_price _)
    //加载天猫数据
    sqlContext.read.orc(s"${source_result}").createOrReplaceTempView("source")

    val source_data = sqlContext.sql(
      s"""
         |select
         |*,
         |cast(salesAmount as double)/cast(sellCount as double) as price_avg
         |from source
       """.stripMargin)
      .withColumnRenamed("priceText","priceText_v1")
      .withColumnRenamed("price_avg","priceText")

    //.selectExpr("good_id","sellCount","priceText","salesAmount","is_newProduct")


    source_data.createOrReplaceTempView("mid1")

    //加载是否是新品集合
   /* val newProduct_data = sqlContext.
      read.json(s"${a_aa_amainpackage.a_o2odata_deal.config.config.is_newProduct_path}")
      .dropDuplicates("good_id")
    newProduct_data.createOrReplaceTempView("newProduct_data")

    //处理是否是新品标签
    val mid1 = sqlContext.sql(
      s"""
         |select
         |t1.*,
         |case
         |when t2.good_id is null then false
         |else t2.is_newProduct
         |end as is_newProducts
         |from source_data t1
         |left join
         |newProduct_data t2
         |on t1.good_id=t2.good_id
       """.stripMargin).drop("is_newProduct").withColumnRenamed("is_newProducts","is_newProduct")
    mid1.createOrReplaceTempView("mid1")

    mid1.agg(count("*"),sum("sellCount"),sum("salesAmount")).show(false)
*/


    //取出下架商品
    val xiajia_data = sqlContext.read.json(a_aa_amainpackage.a_o2odata_deal.config.config.source_obs_data_path)
      .where("is_onsell='false'").selectExpr("good_id")
    xiajia_data.createOrReplaceTempView("xiajia_data")

    //取出数据中下架的商品
    sqlContext.sql(
      s"""
         |select
         |t1.good_id
         |from mid1 t1
         |inner join
         |xiajia_data t2
         |on t1.good_id=t2.good_id
       """.stripMargin).createOrReplaceTempView("xj_data")

    //加载近三个月数据
    sqlContext.read.orc(s"s3a://dws-data/g_data/${years}/${lastmonth}/tmall/")
      .selectExpr("good_id","salesAmount","sellCount")
      .createOrReplaceTempView("data11")

    val data11 = sqlContext.sql(
      s"""
         |select
         |good_id,
         |cast(salesAmount as double)/cast(sellCount as double) as priceText
         |from data11
       """.stripMargin).selectExpr("good_id","priceText")

    sqlContext.read.orc(s"s3a://dws-data/g_data/${last_years}/12/tmall/").selectExpr("good_id","salesAmount","sellCount")
      .createOrReplaceTempView("data10")

    val data10 = sqlContext.sql(
      s"""
         |select
         |good_id,
         |cast(salesAmount as double)/cast(sellCount as double) as priceText
         |from data10
       """.stripMargin).selectExpr("good_id","priceText")

    sqlContext.read.orc(s"s3a://dws-data/g_data/${years}/${lastlastmonth}/tmall/").selectExpr("good_id","salesAmount","sellCount")
      .createOrReplaceTempView("data9")

    val data9 = sqlContext.sql(
      s"""
         |select
         |good_id,
         |cast(salesAmount as double)/cast(sellCount as double) as priceText
         |from data9
       """.stripMargin).selectExpr("good_id","priceText")

    val th_data = data9.unionAll(data10).unionAll(data11)
    th_data.createOrReplaceTempView("th_data")
    sqlContext.sql(
      s"""
         |select
         |good_id,
         |priceText
         |from
         |(select
         |*,
         |row_number() over(partition by good_id order by cast(priceText as double) asc) as rank
         |from th_data)
         |where rank=1
       """.stripMargin).createOrReplaceTempView("hsitory_data")

    sqlContext.sql(
      s"""
         |select
         |t1.good_id,
         |t2.priceText
         |from xj_data t1
         |inner join
         |hsitory_data t2
         |on t1.good_id=t2.good_id
       """.stripMargin).dropDuplicates("good_id").createOrReplaceTempView("price_data")

    val mid2 = sqlContext.sql(
      s"""
         |select
         |*,
         |cast(price as double)*cast(sellCount as bigint) as sales
         |from
         |(select
         |t1.*,
         |case
         |when t2.good_id is null then t1.priceText
         |else min_price(t1.priceText,t2.priceText)
         |end as price
         |from mid1 t1
         |left join
         |price_data t2
         |on t1.good_id=t2.good_id)
       """.stripMargin).drop("priceText","salesAmount")
      .withColumnRenamed("priceText_v1","priceText")
      .withColumnRenamed("sales","salesAmount")
    mid2.createOrReplaceTempView("mid2")

    //mid2.agg(count("*"),sum("sellCount"),sum("salesAmount")).show(false)

    mid2
  }

  def min_price(price1:String,price2:String):String={
    var p = 0.0
    val p1 = price1.toDouble
    val p2 = price2.toDouble
    if (p1<=p2){
      p = p1
    }else{
      p = p2
    }
    p.toString
  }

  def handle_array(liveId_collect:Seq[String],anchorId_collect:Seq[String],nick_collect: Seq[String],
                   viewerCount_collect:Seq[String],liveTime_collect:Seq[String]):String={
    val array = new JSONArray();
    for (i <- 0 to anchorId_collect.length-1){
      val jsonStr = "{" +
        "\"liveId\":"+"\""+liveId_collect(i)+"\"," +
        "\"anchorId\":"+"\""+anchorId_collect(i)+"\"," +
        "\"nick\":"+"\""+nick_collect(i)+"\"," +
        "\"viewCount\":"+"\""+viewerCount_collect(i)+"\"," +
        "\"liveTime\":"+"\""+liveTime_collect(i)+"\"}"
      val jsonObject = JSON.parseObject(jsonStr)
      array.add(jsonObject)
    }
    array.toString
  }


  def tranTimeToString(timestamp:String) :String={
    val tm = timestamp+"000"
    val fm = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")
    val time = fm.format(new Date(tm.toLong))
    time
  }
}
